package com.example.demo.config.listener;

import com.example.demo.config.consumer.RocketMQConsumerProperties;
import java.util.List;
import javax.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.context.annotation.Configuration;

/**
 * @author yongen
 * @description: 抽象消费消息监听类
 * @date 2022/2/9 2:52 PM
 */
@Configuration
@Slf4j
public abstract class AbstractConsumerListener {

  @Resource
  private RocketMQConsumerProperties rocketMQConsumerProperties;

  /**
   * 开启消费注册
   *
   * @param topic
   * @param tags  支持多个tag, 如 tag1 || tag2 || tag3
   * @throws MQClientException
   */
  public void listener(String topic, String tags) throws MQClientException {
    log.info("开启【{}】：【{}】 消费者....", topic, tags);
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(rocketMQConsumerProperties.getGroup());
    consumer.setNamesrvAddr(rocketMQConsumerProperties.getNamesrvAddr());
    consumer.subscribe(topic, tags);
    // 开启内部类实现监听
    consumer.registerMessageListener(new MessageListenerConcurrently() {
      @Override
      public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        return AbstractConsumerListener.this.onMessage(msgs);
      }
    });
    consumer.start();
  }

  /**
   * 处理body的业务
   *
   * @param msgs
   * @return ConsumeConcurrentlyStatus
   */
  public abstract ConsumeConcurrentlyStatus onMessage(List<MessageExt> msgs);

}
